package com.atguigu.flink.chapter05.Sink;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/8/10 8:59
 */
public class Flink03_ES {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> socketDS = env.socketTextStream("localhost", 9999);

        SingleOutputStreamOperator<WaterSensor> sensorDS = socketDS.map((MapFunction<String, WaterSensor>) value -> {
            String[] datas = value.split(",");
            return new WaterSensor(
                    datas[0],
                    Long.parseLong(datas[1]),
                    Integer.parseInt(datas[2])
            );
        });

        List<HttpHost> httpHosts = Arrays.asList(
                new HttpHost("hadoop1", 9200),
                new HttpHost("hadoop2", 9200),
                new HttpHost("hadoop3", 9200)
        );

        //TODO Sink - ES
        ElasticsearchSink.Builder<WaterSensor> esSink = new ElasticsearchSink.Builder<>(
                httpHosts,
                new ElasticsearchSinkFunction<WaterSensor>() {
                    @Override
                    public void process(WaterSensor element, RuntimeContext ctx, RequestIndexer indexer) {
                        HashMap<String, String> dataMap = new HashMap<>();
                        dataMap.put("data", element.toString());
                        IndexRequest indexRequest = Requests.indexRequest()
                                .index("flink210323")
                                .type("_doc")
                                .source(dataMap);

                        indexer.add(indexRequest);
                    }
                });

        // TODO 为了 教学演示，设为1。  生产环境千万不要设为 1 ！！！！！！！！！！！！！
        esSink.setBulkFlushMaxActions(1);

        sensorDS.addSink(esSink.build());

        env.execute();
    }
}
/*
http://hadoop1:9200/_cat/indices?v
http://hadoop1:9200/flink210323/_search
 */